/****************************************************************
 * Licensed to the Apache Software Foundation (ASF) under one   *
 * or more contributor license agreements.  See the NOTICE file *
 * distributed with this work for additional information        *
 * regarding copyright ownership.  The ASF licenses this file   *
 * to you under the Apache License, Version 2.0 (the            *
 * "License"); you may not use this file except in compliance   *
 * with the License.  You may obtain a copy of the License at   *
 *                                                              *
 *   http://www.apache.org/licenses/LICENSE-2.0                 *
 *                                                              *
 * Unless required by applicable law or agreed to in writing,   *
 * software distributed under the License is distributed on an  *
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY       *
 * KIND, either express or implied.  See the License for the    *
 * specific language governing permissions and limitations      *
 * under the License.                                           *
 ****************************************************************/
package org.apache.james.mailbox.cassandra.io;

import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransportException;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ConsistencyLevel;

/**
 * Return an InputStream which retrieve columns from a row which stores chunk of
 * data. See also {@link ChunkOutputStream}
 *
 * This implementation is not thread-safe!
 * 
 * Bsed on Hector implementation for Cassandra.
 * https://github.com/rantav/hector/blob/master/core/src/main/java/me/prettyprint/cassandra/io/ChunkInputStream.java
 */
public class ChunkInputStream extends InputStream {

    //private final Configuration conf;
      
    private final String cf;
    private final String cn;
    private final String key;
    private byte[] chunk;
    private final String keyspace;
    private int pos;
    private long chunkPos = 1;
    
    

    public ChunkInputStream(String keyspace,String cf, String cn, String key) {
        this.key = key;
        this.cn=cn;
        this.cf = cf;
        this.keyspace=keyspace;
    }

    public ChunkInputStream(ByteBuffer keyspace,ByteBuffer cf, ByteBuffer cn, ByteBuffer key) throws CharacterCodingException {
        
        this(ByteBufferUtil.string(keyspace),ByteBufferUtil.string(cf),ByteBufferUtil.string(cn),ByteBufferUtil.string(key));
    }
    
    
     /**
     * Fetch the next chunk.
     *
     * @return exists if there was a chunk to fetch.
     * @throws IOException
     */
    private boolean fetchChunk() throws IOException, TTransportException, InvalidRequestException, TException, UnavailableException, TimedOutException, NotFoundException {
                
             //connection:
            
            TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
            TProtocol protocol = new TBinaryProtocol(transport);
            transport.open();
            Cassandra.Client client = new Cassandra.Client(protocol);
            client.set_keyspace(keyspace);
                   	
            ColumnPath path = new ColumnPath(cf);
 	    path.setColumn(ByteBufferUtil.bytes(cn));
	
             ColumnOrSuperColumn result = client.get(ByteBufferUtil.bytes(key),path,ConsistencyLevel.ONE);
 	     Column cname =  result.getColumn();
             
             if(cname != null)
             {
                 
                 chunk=cname.getValue();
                 chunkPos++;
                 pos = 0;
                 return true;
             }   
             else
                 return false;
             
        }
        
    

    
    @Override
    public int read() throws IOException {
        
    if (chunk == null || pos + 1 == chunk.length) {
            try {
                if (!fetchChunk()) {
                    return -1;
                }
            } catch (TTransportException ex) {
                Logger.getLogger(ChunkInputStream.class.getName()).log(Level.SEVERE, null, ex);
            } catch (InvalidRequestException ex) {
                Logger.getLogger(ChunkInputStream.class.getName()).log(Level.SEVERE, null, ex);
            } catch (TException ex) {
                Logger.getLogger(ChunkInputStream.class.getName()).log(Level.SEVERE, null, ex);
            } catch (UnavailableException ex) {
                Logger.getLogger(ChunkInputStream.class.getName()).log(Level.SEVERE, null, ex);
            } catch (TimedOutException ex) {
                Logger.getLogger(ChunkInputStream.class.getName()).log(Level.SEVERE, null, ex);
            } catch (NotFoundException ex) {
                Logger.getLogger(ChunkInputStream.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
        return chunk[pos++];
    }
    
    
}